package com.luboyan.CBillbill.service;

import com.alibaba.fastjson.JSONObject;
import com.luboyan.CBillbill.dao.file.Danmu;
import com.luboyan.CBillbill.service.file.DanmuService;
import com.luboyan.CBillbill.util.JwtHelper;
import com.luboyan.CBillbill.util.RocketMQUtil;
import com.mysql.cj.util.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import static com.luboyan.CBillbill.util.enums.UserInfoEnum.TOPIC_DANMUS;

/**
 * 处理webSocket相关服务
 * 使用ServerEndpoint注册创建WebSocket类后，该类是由webSocket容器管理而不是由
 * spring框架的IOC容器管理，每个WebSocket客户端连接时都会创建一个新的实例，
 * 因为@ServerEndpoint标注的类并不是由Spring管理的，而是WebSocket运行时管理的。
 * 因此，每次有新的WebSocket连接，底层容器都会创建WebSocketService的新实例。
 * 在这种情况下，由Spring管理的Bean（比如redisTemplate）需要被注入到由WebSocket容器管理的多个WebSocketService实例中，
 * 这里就出现了问题。Spring容器无法自动地将redisTemplate Bean注入到每个新创建的WebSocketService实例中，
 * 因为这些实例不是由Spring容器直接管理的。
 */
@Component
//@ServerEndpoint("/ws-server") //标识为WebSocket相关服务类
@ServerEndpoint("/imserver/{token}") //标识为WebSocket相关服务类
public class WebSocketService {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    //java提供的原子性操作类，为了保证线程安全，代表在线人数
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    //java提供的原子性操作类，为了保证线程安全,为每个客户端分配一个websocket服务
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    private Session session;

    private String sessionId;

    private Long userId;

    //获取到IOC容器
    private static ApplicationContext APPLICATIONCONTEXT;

    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketService.APPLICATIONCONTEXT = applicationContext;
    }


    /**
     * 连接成功后调用该方法
     */
    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token) {
        //使用try是因为游客模式也可以看视频。不一定会有userId
        try {
            this.userId = JwtHelper.verifyToken(token);
        } catch (Exception ignored) {
        }
        this.sessionId = session.getId();
        this.session = session;
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        } else {
            WEBSOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement();
        }
        logger.info("用户连接成功：" + sessionId + "，当前在线人数为：" + ONLINE_COUNT.get());
        //连接成功后返回给前端状态
        try {
            this.sendMessage("600");
        } catch (Exception e) {
            logger.error("连接异常！");
        }
    }

    /**
     * 刷新或关闭页面时会调用onClose方法
     */
    @OnClose
    public void closeConnection() {
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出：" + sessionId + "，当前在线人数为：" + ONLINE_COUNT.get());
    }

    /**
     * 当前端有消息发来时调用这个方法
     */
    @OnMessage
    public void onMessage(String message) {
        logger.info("弹幕信息：会话ID为" + sessionId + ",内容：" + message);
        if (!StringUtils.isNullOrEmpty(message)) {
            try {
                //发送给所有在线客户端信息,采用了MQ+异步发送
                for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
                    WebSocketService webSocketService = entry.getValue();
                    DefaultMQProducer producer = (DefaultMQProducer) APPLICATIONCONTEXT.getBean("danmusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessionId", webSocketService.getSessionId());
                    Message msg = new Message(TOPIC_DANMUS.getParameter(), jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RocketMQUtil.asyncSendMsg(producer, msg);
                }

                if (this.userId != null) {
                    //保存用户发送的弹幕，方便以后快速展示在视频中(还未实现MQ削峰)
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService) APPLICATIONCONTEXT.getBean("DanmuService");
                    danmuService.asyncAddDanmu(danmu);
                    //存入数据库中其实可以设置一个定时任务将每天的弹幕从redis中存入数据库
                    //存入redis
                    danmuService.addDanmusToRedis(danmu);
                }
            } catch (Exception e) {
                logger.error("弹幕接收出错！");
                e.printStackTrace();
            }
        }

    }

    /**
     * 当发生错误时调用这个方法
     */
    @OnError
    public void onError(Throwable throwable) {

    }

    /**
     * 定时发送给前端在线人数（还未实现分视频）
     */
    @Scheduled(fixedRate = 5000) //单位：毫秒
    private void noticeOnlineCount() throws IOException {
        for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
            WebSocketService webSocketService = entry.getValue();
            if (webSocketService.session.isOpen()) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount: ", ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());  
            }
        }
    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message); //将消息方法放入会话中
    }

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }
}
